package com.umeox.moto.mqtt.send;

import android.annotation.SuppressLint;
import android.content.Intent;
import android.text.TextUtils;
import android.util.Log;

import com.umeox.moto.mqtt.BadNetworkException;
import com.umeox.moto.mqtt.BuildConfig;
import com.umeox.moto.mqtt.MqttDefaultCallback;
import com.umeox.moto.mqtt.MqttNet;
import com.umeox.moto.mqtt.ThreadTools;
import com.umeox.moto.mqtt.Utils;
import com.umeox.moto.mqtt.base.AError;
import com.umeox.moto.mqtt.base.ARequest;
import com.umeox.moto.mqtt.base.AResponse;
import com.umeox.moto.mqtt.base.ASend;
import com.umeox.moto.mqtt.base.IOnCallListener;
import com.umeox.moto.mqtt.event.IOnSubscribeListener;
import com.umeox.moto.mqtt.request.MqttPublishRequest;
import com.umeox.moto.mqtt.request.MqttSubscribeRequest;

import org.eclipse.paho.client.mqttv3.IMqttActionListener;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.IMqttToken;
import org.eclipse.paho.client.mqttv3.MqttAsyncClient;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttMessage;

public class MqttSend extends ASend implements IMqttActionListener, MqttCallback {
    private IOnSubscribeListener subscribeListener = null;

    public MqttSend(ARequest request, IOnCallListener callListener) {
        super(request, callListener);
        setStatus(MqttSendStatus.WAITING_TO_SEND);
    }

    public MqttSend(ARequest request, IOnSubscribeListener listener) {
        super(request, null);
        subscribeListener = listener;
        setStatus(MqttSendStatus.WAITING_TO_SEND);
    }

    @Override
    public MqttSendStatus getStatus() {
        return (MqttSendStatus) this.status;
    }

    public void setStatus(MqttSendStatus status) {
        this.status = status;
    }

    public IOnSubscribeListener getSubscribeListener() {
        return subscribeListener;
    }

    @Override
    public void onSuccess(IMqttToken asyncActionToken) {
        if (request instanceof MqttSubscribeRequest) {
            setStatus(MqttSendStatus.COMPLETED);
            boolean successed = true;

            try {
                int suback = asyncActionToken.getGrantedQos()[0];
                if (suback == 0x80) {
                    successed = false;
                }
            } catch (Exception e) {
                if (BuildConfig.DEBUG) {
                    Log.d("MqttSend", "Exception:" + e.getMessage());
                }
            }

            if (subscribeListener != null) {
                if (subscribeListener.needUISafety()) {
                    ThreadTools.runOnUiThread(new CallbackOnUiThreadTask(this, (successed ? InternalMqttRequestState.SUBSCRIBE_SUCCESSED : InternalMqttRequestState.SUBSCRIBE_FAILED), null));
                } else if (successed) {
                    subscribeListener.onSuccess(((MqttSubscribeRequest) request).topic);
                } else {
                    AError error = new AError();
                    error.setCode(AError.AKErrorServerBusinessError);
                    error.setMsg("subACK Failure");
                    subscribeListener.onFailed(((MqttSubscribeRequest) request).topic, error);
                }
            }
        } else if (request instanceof MqttPublishRequest) {
            MqttPublishRequest publishRequest = (MqttPublishRequest) request;
            if (BuildConfig.DEBUG) {
                Log.e("MqttSend", "PublishRequest, onSuccess, topic = " + publishRequest.topic + ",payloadObj:" + publishRequest.payloadObj);
            }
            if (!publishRequest.isRPC) {
                setStatus(MqttSendStatus.COMPLETED);
//                if (onCallListener != null) {
//                    if (onCallListener.needUISafety()) {
//                        ThreadTools.runOnUiThread(new CallbackOnUiThreadTask(this, InternalMqttRequestState.PUBLISH_SUCCESSED, null));
//                    } else {
//                        onCallListener.onSuccess(request, response);
//                    }
//                }
            } else if (status == MqttSendStatus.WAITING_TO_SUB_REPLY) {
                setStatus(MqttSendStatus.SUB_REPLYED);
                MqttSendExecutor.createExecutor().asyncSend(this);
            } else if (status == MqttSendStatus.WAITING_TO_PUBLISH) {
                setStatus(MqttSendStatus.PUBLISHED);
            }
        }
    }

    @Override
    public void onFailure(IMqttToken asyncActionToken, Throwable exception) {
        String errorMsg = null != exception ? exception.getMessage() : "MqttNet send failed: unknown error";
        setStatus(MqttSendStatus.COMPLETED);
        InternalMqttRequestState state;
        AError aError;
        if (request instanceof MqttSubscribeRequest) {
            if (subscribeListener != null) {
                state = InternalMqttRequestState.SUBSCRIBE_FAILED;
                if (exception instanceof BadNetworkException) {
                    state = InternalMqttRequestState.SUBSCRIBE_FAILED_INVOKENET;
                }

                if (subscribeListener.needUISafety()) {
                    ThreadTools.runOnUiThread(new CallbackOnUiThreadTask(this, state, errorMsg));
                } else if (state == InternalMqttRequestState.SUBSCRIBE_FAILED_INVOKENET) {
                    aError = new AError();
                    aError.setCode(AError.AKErrorInvokeNetError);
                    subscribeListener.onFailed(((MqttSubscribeRequest) request).topic, aError);
                } else {
                    aError = new AError();
                    aError.setCode(AError.AKErrorUnknownError);
                    aError.setMsg(errorMsg);
                    subscribeListener.onFailed(((MqttSubscribeRequest) request).topic, aError);
                }
            }
        } else if (request instanceof MqttPublishRequest && onCallListener != null) {
            MqttAsyncClient asyncClient = MqttNet.getInstance().getMqttClient();
            if(asyncClient != null){
                asyncClient.setCallback(new MqttDefaultCallback());
            }
            state = InternalMqttRequestState.PUBLISH_FAILED;
            if (exception instanceof BadNetworkException) {
                state = InternalMqttRequestState.PUBLISH_FAILED_INVOKENET;
            }

            if (onCallListener.needUISafety()) {
                ThreadTools.runOnUiThread(new CallbackOnUiThreadTask(this, state, errorMsg));
            } else if (state == InternalMqttRequestState.PUBLISH_FAILED_INVOKENET) {
                aError = new AError();
                aError.setCode(AError.AKErrorInvokeNetError);
                onCallListener.onFailed(request, aError);
            } else {
                aError = new AError();
                aError.setCode(AError.AKErrorUnknownError);
                aError.setMsg(errorMsg);
                onCallListener.onFailed(request, aError);
            }
        }
    }

    @Override
    public void connectionLost(Throwable cause) {

    }

    @Override
    public void messageArrived(String topic, MqttMessage message) {
        if (BuildConfig.DEBUG) {
            Log.e("MqttSend", "messageArrived(), topic =" + topic + " msg =" + message.toString());
        }
        boolean isMsgMatched = false;
        if (request instanceof MqttPublishRequest) {
            MqttPublishRequest publishRequest = (MqttPublishRequest) request;
            String requestId = Utils.parseRequestId(topic);
            if (BuildConfig.DEBUG) {
                Log.e("MqttSend", "publishRequest =" + publishRequest.toString() + ",requestId=" + requestId);
            }
            if ((status == MqttSendStatus.PUBLISHED || status == MqttSendStatus.COMPLETED)
                    && topic.equals(publishRequest.replyTopic)
                    && !TextUtils.isEmpty(publishRequest.requestId)
                    && publishRequest.requestId.equals(requestId)) {
                if (BuildConfig.DEBUG) {
                    Log.d("MqttSend", "messageArrived(), match!");
                }
                isMsgMatched = true;
                setStatus(MqttSendStatus.COMPLETED);
                if (response == null) {
                    response = new AResponse();
                }
                response.data = message.toString();
                if (onCallListener != null) {
                    if (onCallListener.needUISafety()) {
                        ThreadTools.runOnUiThread(new CallbackOnUiThreadTask(this, InternalMqttRequestState.PUBLISH_SUCCESSED, null));
                    } else {
                        onCallListener.onSuccess(request, response);
                    }
                }
                MqttAsyncClient asyncClient = MqttNet.getInstance().getMqttClient();
                if(asyncClient != null){
                    asyncClient.setCallback(new MqttDefaultCallback());
                }
            }
        }

        if(!isMsgMatched){
            //遗漏没处理的消息就直接发送出去处理
            sendMqttMsgBroadCast(topic,message.getPayload());
            MqttAsyncClient asyncClient = MqttNet.getInstance().getMqttClient();
            if(asyncClient != null){
                asyncClient.setCallback(new MqttDefaultCallback());
            }
        }
    }

    public static final int FLAG_RECEIVER_INCLUDE_BACKGROUND = 0x01000000;
    @SuppressLint("WrongConstant")
    private void sendMqttMsgBroadCast(String topic, byte[] mqttMessage) {
        Intent intent = new Intent();
        intent.setAction("com.umeox.watch.intent.action.MQTT_MSG");
        intent.putExtra("topic",topic);
        intent.putExtra("mqttMessage",mqttMessage);
        intent.addFlags(Intent.FLAG_INCLUDE_STOPPED_PACKAGES | Intent.FLAG_RECEIVER_FOREGROUND | FLAG_RECEIVER_INCLUDE_BACKGROUND);
        MqttNet.getInstance().getContext().sendBroadcast(intent);
    }

    @Override
    public void deliveryComplete(IMqttDeliveryToken token) {

    }

    private enum InternalMqttRequestState {
        SUBSCRIBE_SUCCESSED,
        SUBSCRIBE_FAILED,
        SUBSCRIBE_FAILED_INVOKENET,
        PUBLISH_SUCCESSED,
        PUBLISH_FAILED,
        PUBLISH_FAILED_INVOKENET;
    }

    private class CallbackOnUiThreadTask implements Runnable {
        private MqttSend send;
        private InternalMqttRequestState requestState;
        private String errorMesg;

        private CallbackOnUiThreadTask(MqttSend send, InternalMqttRequestState state, String msg) {
            this.send = send;
            this.requestState = state;
            this.errorMesg = msg;
        }

        @Override
        public void run() {
            if (send != null) {
                switch (requestState) {
                    case SUBSCRIBE_SUCCESSED:
                        if (send.getSubscribeListener() == null) {
                            return;
                        }
                        send.getSubscribeListener().onSuccess(((MqttSubscribeRequest) send.getRequest()).topic);
                        break;
                    case SUBSCRIBE_FAILED:
                    case SUBSCRIBE_FAILED_INVOKENET:
                        if (send.getSubscribeListener() == null) {
                            return;
                        }
                        AError error = new AError();
                        if (requestState == InternalMqttRequestState.PUBLISH_FAILED_INVOKENET) {
                            error.setCode(AError.AKErrorInvokeNetError);
                        } else {
                            error.setCode(AError.AKErrorUnknownError);
                        }
                        error.setMsg(errorMesg);
                        send.getSubscribeListener().onFailed(((MqttSubscribeRequest) send.getRequest()).topic, error);
                        break;
                    case PUBLISH_SUCCESSED:
                        if (send.getOnCallListener() == null) {
                            return;
                        }
                        send.getOnCallListener().onSuccess(send.getRequest(), send.getResponse());
                        break;
                    case PUBLISH_FAILED:
                    case PUBLISH_FAILED_INVOKENET:
                        if (send.getOnCallListener() == null) {
                            return;
                        }
                        error = new AError();
                        if (requestState == InternalMqttRequestState.SUBSCRIBE_FAILED_INVOKENET) {
                            error.setCode(AError.AKErrorInvokeNetError);
                        } else {
                            error.setCode(AError.AKErrorUnknownError);
                        }
                        error.setMsg(errorMesg);
                        send.getOnCallListener().onFailed(send.getRequest(), error);
                    default:
                        break;
                }
            }
        }
    }
}
